**CONNECTION** _connection_name_ | The name of the Kafka connection to use in the source. For details on creating connections, check the [`CREATE CONNECTION`](/sql/create-connection/#kafka) documentation page.
**TOPIC** _topic_ | The Kafka topic you want to subscribe to.
**INCLUDE KEY** | Include a column containing the Kafka message key. If the key is encoded using a format that includes schemas the column will take its name from the schema. For unnamed formats (e.g. `TEXT`), the column will be named `key`. The column can be renamed with the optional **AS** *name* statement.
**INCLUDE PARTITION** | Include a `partition` column containing the Kafka message partition. The column can be renamed with the optional **AS** *name* clause. *New in v0.12.0.*
**INCLUDE OFFSET** | Include an `offset` column containing the Kafka message offset. The column can be renamed with the optional **AS** *name* clause. *New in v0.12.0.*
**INCLUDE TIMESTAMP** | Include a `timestamp` column containing the Kafka message timestamp. The column can be renamed with the optional **AS** *name* clause. *New in v0.12.0.* <br><br>Note that the timestamp of a Kafka message depends on how the topic and its producers are configured. See the [Confluent documentation](https://docs.confluent.io/3.0.0/streams/concepts.html?#time) for details.
**PRIMARY KEY (** _col_list_ **) NOT ENFORCED** | Declare a set of columns as a primary key. For more information, see [`Defining primary keys`](#defining-primary-keys).
